package cn.linjianhui.flink.sample.util;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Utils {
    public static StreamExecutionEnvironment getStreamExecutionEnvironment() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10 * 60 * 1000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(30 * 1000);
        // 失败重启，重启3次，间隔1分钟
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, org.apache.flink.api.common.time.Time.minutes(1)));
        env.setParallelism(1);

        return env;
    }
}
